package cn.kgc.gmall.app.dws;

import cn.kgc.gmall.bean.ProvinceStats;
import cn.kgc.gmall.utils.ClickhouseUtil;
import cn.kgc.gmall.utils.MyKafkaUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {

        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        /*
        //CK相关设置
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        StateBackend fsStateBackend = new FsStateBackend(
                "hdfs://hadoop202:8020/gmall/flink/checkpoint/ProvinceStatsSqlApp");
        env.setStateBackend(fsStateBackend);
        System.setProperty("HADOOP_USER_NAME","atkgc");
        */
        //TODO 1.定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        String topicName = "dwm_order_wide";
        String groupId = "province_stats";
        // 去连接流数据 转化表   编写sql语句完成统计
        // 创建动态表  消费kafka中的order_wide主题  获取订单和地区相关信息 设置水位线
        tableEnv.executeSql("create table ORDERWIDE (" +
                "province_id BIGINT,"+
                "province_name STRING,"+
                "province_area_code STRING,"+
                "province_iso_code STRING,"+
                "province_3166_2_code STRING,"+
                "order_id STRING,"+
                "split_total_amount DOUBLE,"+
                "create_time STRING,"+
                "rowtime as TO_TIMESTAMP(create_time),"+
                "WATERMARK FOR  rowtime  AS rowtime"+
                ") WITH("+ MyKafkaUtils.getKafkaDDL(topicName,groupId) +")");
        // sql:1.开窗滚动开窗10秒 统计订单id 去重 统计订单金额 分组根据地区和窗口分组
        Table table = tableEnv.sqlQuery("select " +
                " DATE_FORMAT( TUMBLE_START(rowtime,interval '10' second),'yyyy-MM-dd HH:mm:ss') stt, "+
                " DATE_FORMAT( TUMBLE_END(rowtime,interval '10' second),'yyyy-MM-dd HH:mm:ss') edt, "+
                " province_id,province_name,province_area_code area_code ,province_iso_code iso_code,province_3166_2_code iso_3166_2, "+
                " COUNT(DISTINCT order_id) order_count,SUM(split_total_amount) order_amount, "+
                " UNIX_TIMESTAMP() * 1000 ts "+
                " from ORDERWIDE " +
                " group by TUMBLE(rowtime,interval '10' second)," +
                "province_id,province_name,province_area_code,province_iso_code,province_3166_2_code ");

        // 转化为流
        DataStream<ProvinceStats> provinceStatsDataStream = tableEnv.toAppendStream(table, ProvinceStats.class);
        // 写入到clickHouse中
        provinceStatsDataStream.addSink(
                ClickhouseUtil.getJdbcSink(
                        "insert into province_stats_2022 values(?,?,?,?,?,?,?,?,?,?)"
                )
        );
        provinceStatsDataStream.print();
        env.execute();
    }
}
